/*
package com.zlx.rulemk.marketing.functions;

import com.alibaba.fastjson.JSON;
import com.zlx.rulemk.marketing.beans.*;
import com.zlx.rulemk.marketing.metrics.RuleIsMatchAvgTimeGauge;
import com.zlx.rulemk.marketing.service.RulelMatchServiceImpl;
import com.zlx.rulemk.marketing.utils.FlinkStateDescUtils;
import lombok.extern.slf4j.Slf4j;
import org.apache.flink.api.common.state.BroadcastState;
import org.apache.flink.api.common.state.ListState;
import org.apache.flink.api.common.state.MapState;
import org.apache.flink.api.common.state.ReadOnlyBroadcastState;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.metrics.Counter;
import org.apache.flink.shaded.guava18.com.google.common.base.Stopwatch;
import org.apache.flink.streaming.api.functions.co.KeyedBroadcastProcessFunction;
import org.apache.flink.util.Collector;
import org.kie.api.io.ResourceType;
import org.kie.api.runtime.KieSession;
import org.kie.internal.utils.KieHelper;

import java.util.List;
import java.util.concurrent.TimeUnit;

*/
/**
 * @Author: zhanglingxing
 * @CreateTime: 2022-07-11
 * @Description: 规则匹配核心类
 * @Version: 1.0
 *//*

@Slf4j
public class RuleMatchKeyedProcessFunction extends KeyedBroadcastProcessFunction<String, DynamicKeyedBean,
        CDCLogBean, RuleMatchResult> {

    // flink缓存的明细数据
    transient ListState<EventBean> listState;

    // Flink执行规则对象
    // <ruleName,ruleStateBean>
    // transient MapState<String, RuleStateBean> ruleState;

    // 触发规则的计算类
    RulelMatchServiceImpl rulelMatchServiceImpl;

    //String 是分组的key，不是具体的value，将来获取对应的规则执行
    ListState<Tuple2<String, Long>> timerInfoState;

    Counter rule_calc_timeAmount;
    Counter rule_match_calc_timeAmount;
    Counter rule_not_match_calc_timeAmount;
    Counter rule_calc_cntAmount;
    Counter rule_is_match_calc_cntAmount;
    Counter rule_not_match_calc_cntAmount;

    RuleIsMatchAvgTimeGauge ruleIsMatchAvgTimeGauge;

    @Override
    public void open(Configuration parameters) throws Exception {

        listState = getRuntimeContext().getListState(FlinkStateDescUtils.getEventBeansDesc());

        // ruleState = getRuntimeContext().getMapState(FlinkStateDescUtils.ruleStateDesc2);

        rulelMatchServiceImpl = new RulelMatchServiceImpl(listState);

        // 记录规则定时信息的state
        timerInfoState = getRuntimeContext().getListState(FlinkStateDescUtils.ruleTimerStateDesc);

        rule_calc_timeAmount = getRuntimeContext().getMetricGroup().counter("rule_calc_timeAmount");
        rule_match_calc_timeAmount = getRuntimeContext().getMetricGroup().counter("rule_is_match_calc_timeAmount");
        rule_not_match_calc_timeAmount = getRuntimeContext().getMetricGroup().counter("rule_not_match_calc_timeAmount");
        rule_calc_cntAmount = getRuntimeContext().getMetricGroup().counter("rule_calc_cntAmount");
        rule_is_match_calc_cntAmount = getRuntimeContext().getMetricGroup().counter("rule_is_match_calc_cntAmount");
        rule_not_match_calc_cntAmount = getRuntimeContext().getMetricGroup().counter("rule_not_match_calc_cntAmount");

        ruleIsMatchAvgTimeGauge = getRuntimeContext().getMetricGroup().gauge("计算成立的规则的平均每次计算耗时",  new RuleIsMatchAvgTimeGauge());

    }

    @Override
    public void processElement(DynamicKeyedBean dynamicKeyedBean, ReadOnlyContext ctx,
                               Collector<RuleMatchResult> out) throws Exception {

        // Flinkstate规则对象
        ReadOnlyBroadcastState<String, RuleStateBean> broadcastState = ctx.getBroadcastState(FlinkStateDescUtils.rule_broadcast_state_2);

        // 将日志数据添加到flinkState中
        listState.add(dynamicKeyedBean.getEventBean());

        RuleStateBean ruleStateBean = broadcastState.get(dynamicKeyedBean.getRuleName());

        MarketingRule marketingRule = ruleStateBean.getMarketingRule();

        KieSession kieSession = ruleStateBean.getKieSession();

        */
/**
         * 第一次匹配的时候  timerMatchDroolsFact 为null 先判断【触发条件，画像条件，行为组合条件】如果满足，在注册定时器条件
         *//*

        RuleMatchFact ruleMatchFact = new RuleMatchFact(marketingRule, rulelMatchServiceImpl,
                dynamicKeyedBean.getEventBean(), false,null);

        // 创建stopwatch并开始计时
        Stopwatch stopwatch = Stopwatch.createStarted();

        // 调用drools引擎，执行计算
        kieSession.insert(ruleMatchFact);
        kieSession.fireAllRules();

        // 获取计算结果
        boolean matchResult = ruleMatchFact.isMatchResult();
        long e = System.currentTimeMillis();
        log.debug("结束计算,计算结果：{},耗时{}ms,动态分组的key：{},分组的值：{},日志数据:{}，规则rule_name：{}",
                matchResult,(stopwatch.elapsed(TimeUnit.MILLISECONDS)),
                dynamicKeyedBean.getKeyNames(),dynamicKeyedBean.getKeyValue(),
                dynamicKeyedBean.getEventBean(),marketingRule.getRuleId()
        );

        rule_calc_timeAmount.inc(stopwatch.elapsed(TimeUnit.MILLISECONDS));
        if(matchResult){
            rule_match_calc_timeAmount.inc(stopwatch.elapsed(TimeUnit.MILLISECONDS));
            rule_is_match_calc_cntAmount.inc();

            ruleIsMatchAvgTimeGauge.incRuleIsMatchCount();
            ruleIsMatchAvgTimeGauge.incRuleIsMatchTimeAmount(stopwatch.elapsed(TimeUnit.MILLISECONDS));

        }else{
            rule_not_match_calc_timeAmount.inc(stopwatch.elapsed(TimeUnit.MILLISECONDS));
            rule_not_match_calc_cntAmount.inc();
        }

        // 判断计算结果是否为true, 并继续判断是否有定时条件
        if(matchResult){
            // 再判断这个规则是否是一个带定时条件的规则
            if (marketingRule.isOnTimer()) {  // 是带定时条件的规则
                // 从规则中取出所有定时条件
                List<TimerCondition> timerConditionList = marketingRule.getTimerConditionList();
                //TODO 目前限定一个规则中只有一个时间条件
                TimerCondition timerCondition = timerConditionList.get(0);
                // 注册定时器
                long triggerTime = dynamicKeyedBean.getEventBean().getTimeStamp() + timerCondition.getTimeLate();
                ctx.timerService().registerEventTimeTimer(triggerTime);

                long currentWatermark = ctx.timerService().currentWatermark();

                // 在 定时信息state中进行记录
                timerInfoState.add(Tuple2.of(dynamicKeyedBean.getKeyNames(), triggerTime));

                log.debug("注册定时器规则，动态的key：{},触发时间：{},当前系统的时间:{}，定时规则条件：{}",ctx.getCurrentKey(),
                        triggerTime,currentWatermark,ruleStateBean.getMarketingRule().getTimerConditionList());

            } else {  // 不带定时条件
                // 输出规则成立的结果
                out.collect(new RuleMatchResult(dynamicKeyedBean.getKeyValue(),marketingRule.getRuleId(),dynamicKeyedBean.getEventBean().getTimeStamp(),System.currentTimeMillis()));
            }
        }


    }

    @Override
    public void processBroadcastElement(CDCLogBean cdcLogBean, Context context,
                                        Collector<RuleMatchResult> collector) throws Exception {

        BroadcastState<String, RuleStateBean> broadcastState = context.getBroadcastState(FlinkStateDescUtils.rule_broadcast_state_2);

        // 规则平台上操作的规则的数据
        List<RuleTableRecord> ruleTableRecords = cdcLogBean.getData();
        if(ruleTableRecords == null ) return;

        // 规则平台上的操作的类型
        String operationType = cdcLogBean.getType();

        // 根据不同的操作类型，去操作state中的规则数据
        for (RuleTableRecord ruleTableRecord : ruleTableRecords) {

            String rule_condition_json = ruleTableRecord.getRule_condition_json();
            // 规则条件封装对象
            MarketingRule marketingRule = JSON.parseObject(rule_condition_json, MarketingRule.class);

            String rule_controller_drl = ruleTableRecord.getRule_controller_drl();
            // 规则名称
            String ruleName = ruleTableRecord.getRule_name();

            // 如果是插入操作，或者是 （设置状态为1的更新操作）
            // <rule_name1,ruleStateBean1>

            if ("INSERT".equals(operationType) ||
                    ("UPDATE".equals(operationType) && "1".equals(ruleTableRecord.getRule_status()))) {

                // 往该list中放入一个新的  规则bean对象
                // 往statebean中放kiesession
                KieHelper kieHelper = new KieHelper();
                kieHelper.addContent(rule_controller_drl, ResourceType.DRL);
                KieSession kieSession = kieHelper.build().newKieSession();
                RuleStateBean ruleStateBeanNew = new RuleStateBean(marketingRule, kieSession);
                broadcastState.put(ruleName,ruleStateBeanNew);
                log.debug("【新增/更新】规则 :规则ruleName：{}",ruleName);
            }
            if ("DELETE".equals(operationType) ||
                    ("UPDATE".equals(operationType) && "0".equals(ruleTableRecord.getRule_status()))) {
                broadcastState.remove(ruleName);
                log.debug("删除规则 :规则ruleName：{}",ruleName);
            }

        }


    }


}
*/
